package com.mq.java.test;

import com.mq.common.MqConnectionUtils;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author zhanghaidong
 * @version v1.0
 * @description 消息生产者
 * @date 2019/8/16
 *
 *
 * 交换器
 *
 * 队列
 *
 * 路由键；路由规则
 *
 * 通过路由键把交换器和队列绑定
 *
 */
public class Producer {

    public static void main(String[] args) {

        //================================================================================================
        //
        // 是该由生产者还是消费者创建所需的队列呢?
        //     如果生产者不创建的话，会面临消息进入“黑洞”，消息会丢失
        //     最好双方都去尝试创建队列
        //
        //================================================================================================


        try {
            // 连接到RabbitMQ
            Connection connection = MqConnectionUtils.getConnection();
            /**
             * 获取信道
             *     在一条TCP连接上创建多少条信道是没有限制的
             */
            Channel channel = connection.createChannel();
            System.out.println("信道id = " + channel.getChannelNumber());

            /**
             * 声明交换器(交换机)
             *      1、exchange：交换器名称
             *      2、type：交换器类型
             *      3、durable：是是否持久化,durable设置为true表示持久化,反之是非持久化,持久化的可以将交换器存盘,在服务器重启的时候不会丢失信息
             *      4、autoDelete：是否自动删除，如果没有与之绑定的Queue，直接删除
             *      5、internal：是否内置的，如果为true，只能通过Exchange到Exchange
             *      6、arguments：结构化参数
             *
             *  不存在创建，存在就
             */
            String exchangeName = "direct-exchange";
            channel.exchangeDeclare(exchangeName,
                    BuiltinExchangeType.DIRECT,
                    false,
                    false,
                    false,
                    null);

            /**
             * 交换器和交换器绑定
             *
             * 作为RabbitMQ对AMQP的扩展功能，交换器允许和别的交换器绑定起来。这和队列与交换器对绑定在使用上没有太大对不同。可以将交换器E1和E2绑定起来，在将E2和队列
             * Q绑定起来，之后生产者向E1发送消息，消息在E1上被路由到E2，再由E2路由到Q。
             */
            //channel.exchangeBind("dest-exchange", "src-exchange", "exchange-to-exchange", null);


            /**
             * check if the named exchange exists
             * passive
             *   如果交换器存在返回成功
             *   如果不存在返回错误
             *
             *   exchangeDeclarePassive
             *   主要用来检测相应的交换器是否存在,如果存在就正常返回,不存就是抛出异常404 channel exception,同是channel也会被关闭
             */
            //exchangeDeclarePassive(connection, channel);

            /**
             * 比exchangeDeclare多设置了个nowait参数,
             * 不需要服务器返回,这个方法返回是void,而普通的exchangeDeclare方法返返回值是Exchange.DeclareOk
             * 意思是在客户商端声明一个交换器之后,需要等待服务器返回(服务器会返回Exchange.DeclareOk)
             * exchangeDeclareNoWait在声明一个交换器后,实际上服务器并没有完成交换器的创建,此时客户端接着使用这个交换器,就会发生异常,不建议使用个这方法
             *
             * 目前不知道有什么特殊用处
             */
//            channel.exchangeDeclareNoWait(exchangeName,
//                    BuiltinExchangeType.DIRECT,
//                    false,
//                    false,
//                    false,
//                    null);

            for (int i = 0; i < 10000 ; i++) {
                basicPublish(channel, exchangeName, "Hello World:" + i);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            // 渠道关闭
            channel.close();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }

    private static void basicPublish(Channel channel, String exchangeName, String message) throws IOException {
        /**
         * 发布消息
         *   1、exchange：交换机名称
         *   2、routingKey：路由键
         *   3、mandatory的作用：
         *          当mandatory标志位设置为true时，如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息，那么broker会调用basic.return方法将消息返还给生产者;
         *          当mandatory设置为false时，出现上述情况broker会直接将消息丢弃;通俗的讲，mandatory标志告诉broker代理服务器至少将消息route到一个队列中，否则就将消息return给发送者;
         *   4、immediate RabbitMQ不在支持该标志了 【该参数从RabbitMQ 3.0版本开始已经废弃掉】
         *   5、props：消息属性字段，比如消息头部信息等等
         *   6、body：消息主体部分
         *
         *
         *   消息包含两部分：有效载荷（payload）和标签（label）
         */
        channel.basicPublish(exchangeName,
                "direct-routingKey",
                false,
                false,
                MessageProperties.TEXT_PLAIN,
                message.getBytes());
    }

    /**
     * 检查交换器是否存在
     * @param connection
     * @param channel
     * @throws IOException
     * @throws TimeoutException
     */
    private static void exchangeDeclarePassive(Connection connection, Channel channel) throws IOException, TimeoutException {
        try {
            AMQP.Exchange.DeclareOk declareOk = channel.exchangeDeclarePassive("direct-exchange1");
            // #method<exchange.declare-ok>()
            System.out.println(declareOk.toString());
        } catch (Exception e) {
            // channel is already closed due to channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'direct-exchange1' in vhost '/', class-id=40, method-id=10)
            System.out.println("交换器不存在");
        }
    }
}
